package com.enlorenz.core.channel.socket.imp.longconnection;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;

import com.enlorenz.common.util.CommonUtils;
import com.enlorenz.common.util.SocketUtils;
import com.enlorenz.common.util.StreamUtils;
import com.enlorenz.common.util.TimeUtils;
import com.enlorenz.core.channel.IChannelEngine;
import com.enlorenz.core.channel.socket.common.dstroute.category.SocketDstRoute;
import com.enlorenz.core.channel.socket.common.reader.ISocketReader;
import com.enlorenz.core.channel.socket.common.translate.DataMessageTransformer;
import com.enlorenz.core.channel.socket.common.translate.IMessageBytesTransformer;
import com.enlorenz.core.channel.socket.common.util.RecvMessageOperator;
import com.enlorenz.core.channel.socket.common.writer.ISocketWriter;
import com.enlorenz.core.engine.AbstractEngine;
import com.enlorenz.core.engine.EngineUtils;
import com.enlorenz.core.engine.IEngine;
import com.enlorenz.core.engine.exception.EngineControlException;
import com.enlorenz.core.engine.imp.thread.quick.AbstractQuickControlThread;
import com.enlorenz.core.queue.proc.AbstractObjectCycleThread;
import com.enlorenz.core.queue.transfer.MessageQueue;
import com.enlorenz.core.queue.transfer.imp.WaitMessageQueue;

/**
 * 长连接客户端
 * 用于主动发起链接请求并保持链接的通道
 * 接收到的消息通过放入recvQueue队列
 * 需要发送消息则放入sendQueue，此通道会自动获取并发送到链接的服务端
 * 需要设置的参数：
 * ip:ip地址字符串
 * port:端口
 * recvQueue：接收消息的队列
 * sendQueue：发送消息的队列
 * socketWriter:socket输出器
 * socketReader:socket读取器
 * messageBytesTransformer:消息转换器,如果没有传入值，则使用默认的转换器
 * @author lcb
 */
public class LongClientSocketChannel extends AbstractEngine implements IChannelEngine{
	//IP地址
	private String ip;
	//端口
	private Integer port;
	//socket链接
	private Socket socket;
	//socket链接对象
	private SocketDstRoute socketConnObject=new SocketDstRoute();
	//socket输入流
	private InputStream inputStream;
	//socket输出流
	private OutputStream outputStream;
	//接收消息的队列
	protected MessageQueue recvQueue;
	//发送消息的队列
	protected WaitMessageQueue sendQueue;
	//是否输出错误
	private boolean isWriteError=false;
	//socket输出器
	private ISocketWriter socketWriter;
	//socket读取器
	private ISocketReader socketReader;
	//socket发送线程
	private ISendThread sendThread=new SendThread();
	//socket读取线程
	private IRecvThread recvThread=new RecvThread();
	//消息转换器,如果没有传入值，则使用默认的转换器
	private IMessageBytesTransformer messageBytesTransformer
	=new DataMessageTransformer();
	//接收消息操作者
	private RecvMessageOperator recvMessageOperator
	=new RecvMessageOperator();

	/**
	 * 初始化控制线程
	 */
	protected void initEngine() throws EngineControlException{
		stopSendThread();
		
		isWriteError=false;
		sendThread.setChannel(this);
		sendThread.init();
		
		recvThread.setChannel(this);
		recvThread.init();
		
		recvMessageOperator.setRecvQueue(recvQueue);
		
		startSendThread();
	}
	
	/**
	 * 启动Socket服务
	 */
	protected void startEngine() throws EngineControlException {
		try {
			//启动服务
			socketClientStart();
		} catch (Exception e) {
			new EngineControlException(this.getName(),e);
		}
	}
	
	/**
	 * socket服务启动
	 * @throws IOException
	 */
	protected void socketClientStart() throws Exception {
		try {
			socket=new Socket(ip,port);
			socket.setKeepAlive(true);
			socketConnObject.setConnSource(socket);
			outputStream=socket.getOutputStream();
			inputStream=socket.getInputStream();
			//启动接收线程
			startRecvThread();
		} catch (Exception e) {
			close();
			throw e;
		}
		
	}
	
	/**
	 * 关闭打开的链接
	 */
	private void close() {
		//如果socket还在链接，则关闭
		if (SocketUtils.isConnected(socket)) {
			//关闭输入流
			StreamUtils.closeStream(inputStream);
			inputStream=null;
			//关闭输出流
			StreamUtils.closeStream(outputStream);
			outputStream=null;
			//关闭socket链接
			SocketUtils.closeSocket(socket);
			socket=null;
			
			socketConnObject.setConnSource(null);
		}
		//关闭接收线程
		stopRecvThread();
	}
	
	/**
	 * 启动发送线程
	 */
	protected void startSendThread(){
		sendThread.setMessageQueue(sendQueue);
		sendThread.setSocketWriter(socketWriter);
		sendThread.setMessageBytesTransformer(messageBytesTransformer);
		sendThread.start();
	}
	
	/**
	 * 启动接收线程
	 */
	protected void startRecvThread(){
		recvThread.setSocketReader(socketReader);
		recvThread.setMessageBytesTransformer(messageBytesTransformer);
		recvThread.start();
	}
	
	/**
	 * 关闭接收线程
	 */
	protected void stopRecvThread(){
		//关闭接收线程
		EngineUtils.shutdown(recvThread);
	}
	
	
	/**
	 * 关闭发送线程
	 */
	protected void stopSendThread(){
		//关闭发送线程
		EngineUtils.shutdown(sendThread);
	}
	
	/**
	 * 关闭Socket服务
	 */
	@Override
	protected void shutdownEngine() throws EngineControlException {
		//关闭选择器和服务
		close();
	}
	
	/**
	 * Socket是否还存活
	 */
	@Override
	public boolean isAlive() {
		//如果没有打开则没有链接上
		if(!SocketUtils.isConnected(socket)){
			return false;
		}
		//如果是写出错误，那么此线程也无法存活
		if(isWriteError){
			return false;
		}
		
		return true;
	}
	
	/**
	 * Socket发送线程
	 * @author MAC
	 */
	public static interface ISendThread extends IEngine{
		/**
		 * 设置长接连通道
		 * @param channel 长接连通道
		 */
		void setChannel(LongClientSocketChannel channel);
		/**
		 * 设置发送队列
		 * @param messageQueue 发送队列
		 */
		void setMessageQueue(WaitMessageQueue messageQueue);
		/**
		 * 设置socket写出器
		 * @param socketWriter socket写出器
		 */
		void setSocketWriter(ISocketWriter socketWriter);
		/**
		 * 设置消息转换器
		 * @param messageBytesTransformer 消息转换器
		 */
		void setMessageBytesTransformer(
				IMessageBytesTransformer messageBytesTransformer);
	}
	
	/**
	 * 发送线程
	 * @author lcb
	 */
	public class SendThread<T> extends AbstractObjectCycleThread<T> implements ISendThread{
		private LongClientSocketChannel channel;
		private ISocketWriter socketWriter;
		private IMessageBytesTransformer messageBytesTransformer;
		public void setSocketWriter(ISocketWriter socketWriter) {
			this.socketWriter = socketWriter;
		}
		public void setChannel(LongClientSocketChannel channel) {
			this.channel = channel;
		}
		
		public void setMessageBytesTransformer(
				IMessageBytesTransformer messageBytesTransformer) {
			this.messageBytesTransformer = messageBytesTransformer;
		}
		
		@Override
		protected void process(T result) {
			try{
				//将消息转化为字节流
				byte [] data=messageBytesTransformer.message2Bytes(result);
				OutputStream os=channel.getOutputStream();
				if(null == os){
					throw new IOException("socket["+
							LongClientSocketChannel.this.getName()+"]输出流获取为空,请检查socket通道是否建立成功!");
				}
				//写出字节流
				socketWriter.write(os,data);
				channel.setWriteError(false);
				//发送成功消息
				recvMessageOperator.sendWriteStatusMessage(true,result);
			}catch (Exception e) {
				//发送异常消息
				recvMessageOperator.sendWriteStatusMessage(false,result,e);
				
				channel.setWriteError(true);
				//如果出错，则在5秒后重试
				CommonUtils.sleep(TimeUtils.SECOND_MILLI*5);
				throw new RuntimeException(e);
			}
		}

		@Override
		public void setMessageQueue(WaitMessageQueue messageQueue) {
			super.setWaitMessageQueue(messageQueue);
		}
	}
	
	
	
	
	
	
	/**
	 * 接收线程
	 * @author MAC
	 */
	public static interface IRecvThread extends IEngine{
		/**
		 * 设置长接连通道
		 * @param channel 长接连通道
		 */
		void setChannel(LongClientSocketChannel channel);
		/**
		 * 设置socket读取器
		 * @param socketReader socket读取器
		 */
		void setSocketReader(ISocketReader socketReader);
		/**
		 * 设置消息转换器
		 * @param messageBytesTransformer 消息转换器
		 */
		void setMessageBytesTransformer(
				IMessageBytesTransformer messageBytesTransformer);
	}
	
	/**
	 * 发送线程
	 * @author lcb
	 */
	public class RecvThread extends AbstractQuickControlThread implements IRecvThread{
		private LongClientSocketChannel channel;
		private ISocketReader socketReader;
		private IMessageBytesTransformer messageBytesTransformer;
		
		public void setMessageBytesTransformer(
				IMessageBytesTransformer messageBytesTransformer) {
			this.messageBytesTransformer = messageBytesTransformer;
		}
		

		public void setSocketReader(ISocketReader socketReader) {
			this.socketReader = socketReader;
		}

		public void setChannel(LongClientSocketChannel channel) {
			this.channel = channel;
		}

		@Override
		protected void handle() throws Exception {
			socketReader.setInputStream(channel.getInputStream());
			try{
				channel.setWriteError(false);
				//读取字节流
				byte [] data=socketReader.read();
				if(null == data){
					return;
				}
				//将byte转化为内部消息
				Object message=messageBytesTransformer.
					bytes2Message(data);
				//发送接收到的消息
				recvMessageOperator.sendRecvMessage(message);
			}catch (Exception e) {
				channel.setWriteError(true);
				CommonUtils.sleep(TimeUtils.SECOND_MILLI*5);
				throw e;
			}
		}
	}
	/**
	 * 设置IP地址
	 * @param ip IP地址
	 */
	public void setIp(String ip) {
		this.ip = ip;
	}
	/**
	 * 设置端口
	 * @param port 端口
	 */
	public void setPort(Integer port) {
		this.port = port;
	}
	/**
	 * 设置发送队列
	 * @param sendQueue 发送队列
	 */
	public void setSendQueue(WaitMessageQueue sendQueue) {
		this.sendQueue = sendQueue;
	}
	/**
	 * 获取输入流
	 * @return
	 */
	public InputStream getInputStream() {
		return inputStream;
	}
	/**
	 * 获取接收队列
	 * @return 接收队列
	 */
	public MessageQueue getRecvQueue() {
		return recvQueue;
	}
	/**
	 * 获取输出流
	 * @return 输出流
	 */
	public OutputStream getOutputStream() {
		return outputStream;
	}
	/**
	 * 是否写出错误
	 * @return 是否写出错误
	 */
	public boolean isWriteError() {
		return isWriteError;
	}
	/**
	 * 设置是否写出错误
	 * @param isWriteError 是否写出错误
	 */
	public void setWriteError(boolean isWriteError) {
		this.isWriteError = isWriteError;
	}
	
	/**
	 * 设置socket写出器
	 * @param socketWritersocket写出器
	 */
	public void setSocketWriter(ISocketWriter socketWriter) {
		this.socketWriter = socketWriter;
	}
	/**
	 * 获取socket对象
	 * @return
	 */
	public Socket getSocket() {
		return socket;
	}
	/**
	 * 设置socket读取对象
	 * @param socketReader socket读取对象
	 */
	public void setSocketReader(ISocketReader socketReader) {
		this.socketReader = socketReader;
	}
	/**
	 * 设置接收队列
	 * @param recvQueue 接收队列
	 */
	public void setRecvQueue(MessageQueue recvQueue) {
		this.recvQueue = recvQueue;
	}
	/**
	 * 设置发送线程
	 * @param sendThread
	 */
	public void setSendThread(ISendThread sendThread) {
		this.sendThread = sendThread;
	}
	/**
	 * 设置接收线程
	 * @param recvThread
	 */
	public void setRecvThread(IRecvThread recvThread) {
		this.recvThread = recvThread;
	}
	/**
	 * 设置消息转换器
	 * @param messageBytesTransformer
	 */
	public void setMessageBytesTransformer(
			IMessageBytesTransformer messageBytesTransformer) {
		this.messageBytesTransformer = messageBytesTransformer;
	}
	/**
	 * 获取socket链接对象
	 * @return
	 */
	public SocketDstRoute getSocketConnObject() {
		return socketConnObject;
	}
	
}
